// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include <pollux/common/base/exceptions.h>
#include <pollux/dwio/catalog/hive/file_utils.h>

using namespace ::testing;
using namespace kumo::pollux::dwio::catalog::hive;

TEST(FileUtilsTests, MakePartName) {
  std::vector<std::pair<std::string, std::string>> pairs{
      {"ds", "2016-01-01"}, {"FOO", ""}, {"a\nb:c", "a#b=c"}};
  ASSERT_EQ(
      FileUtils::makePartName(pairs, true),
      "ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Dc");
  ASSERT_EQ(
      FileUtils::makePartName(pairs, false),
      "ds=2016-01-01/FOO=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Dc");
}

TEST(FileUtilsTests, ParsePartKeyValues) {
  EXPECT_THROW(
      FileUtils::parsePartKeyValues("ds"), kumo::pollux::PolluxRuntimeError);
  EXPECT_THROW(
      FileUtils::parsePartKeyValues("ds=/ts"),
      kumo::pollux::PolluxRuntimeError);

  try {
    FileUtils::parsePartKeyValues("ts=%xx");
    FAIL()
        << "Expected PolluxException with message 'incomplete percent encode sequence' but none was thrown.";
  } catch (const kumo::pollux::PolluxException& e) {
    EXPECT_THAT(
        e.what(), ::testing::HasSubstr("incomplete percent encode sequence"));
  } catch (...) {
    FAIL() << "Expected PolluxException but caught an unknown exception type.";
  }

  EXPECT_THAT(
      FileUtils::parsePartKeyValues(
          "ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Dc%2F"),
      ElementsAre(
          std::make_pair("ds", "2016-01-01"),
          std::make_pair("foo", "__HIVE_DEFAULT_PARTITION__"),
          std::make_pair("a\nb:c", "a#b=c/")));
}

TEST(FileUtilsTests, ExtractPartitionName) {
  struct TestCase {
   public:
    TestCase(const std::string& filePath, const std::string& partitionName)
        : filePath{filePath}, partitionName{partitionName} {}

    std::string filePath;
    std::string partitionName;
  };

  std::vector<TestCase> testCases{
      {"", ""},
      // identity
      {"ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Ac",
       "ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Ac"},
      // escaped partition name is not a special case.
      {"ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac%3Da%23b%3Ac",
       "ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__"},
      // full path
      {"ws://nobodycares/notimportant/somethingsomething/yaddayadda/ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Ac/part0",
       "ds=2016-01-01/foo=__HIVE_DEFAULT_PARTITION__/a%0Ab%3Ac=a%23b%3Ac"}};

  for (auto testCase : testCases) {
    EXPECT_EQ(
        testCase.partitionName,
        FileUtils::extractPartitionName(testCase.filePath));
  }
}
